package zjs.dc.controller;

import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.SparkSession;

/**
 * @Class:
 * @Author: guoml
 * @Description: 针对企查查t_tm表和中数t_trademarkinfo表进行数据映射开发
 * @Date: 2021/12/7 12:51
 */

public class TTm {
    public static void main(String[] args) {
        SparkSession spark = SparkSession.builder().appName("t_tm").enableHiveSupport().getOrCreate();
        spark.sparkContext().setLogLevel("ERROR");
        Dataset<Row> zs_data = spark.sql("select * from a_dc_ep_ods.zs_t_trademarkinfo");
        long zs_count = zs_data.count();
        if(zs_count>0){

            //获取中数增量数据表中的更新数据（inner join）
            Dataset<Row> update_data = spark.sql("SELECT a.id,\n" +
                    "       a.company_id,\n" +
                    "       d.company_name,\n" +
                    "       a.tm_id,\n" +
                    "       b.MARKNAME,\n" +
                    "       b.MARKCODEKEY,\n" +
                    "       b.APPERSON,\n" +
                    "       a.applicant_en,\n" +
                    "       a.applicant_cn_key_no,\n" +
                    "       case when b.ISINVALID='有效' then '1'" +
                    "            when b.ISINVALID='无效' then '2'" +
                    "            when b.ISINVALID is null then a.status" +
                    "            else '5' end as status,\n" +
                    "       b.UNIONTYPECODE,\n" +
                    "       b.UNIONTYPE,\n" +
                    "       b.APPDATE,\n" +
                    "       a.agent,\n" +
                    "       a.valid_period,\n" +
                    "       a.announcement_issue,\n" +
                    "       a.announcement_date,\n" +
                    "       a.reg_issue,\n" +
                    "       a.reg_date,\n" +
                    "       a.flow_status,\n" +
                    "       a.flow_status_desc,\n" +
                    "       a.flow_items,\n" +
                    "       a.list_group_items,\n" +
                    "       a.address_cn,\n" +
                    "       a.address_en,\n" +
                    "       a.has_image,\n" +
                    "       a.image_url,\n" +
                    "       a.guo_ji_zhu_ce_date,\n" +
                    "       a.hou_qi_zhi_ding_date,\n" +
                    "       a.you_xian_quan_date,\n" +
                    "       a.update_time,\n" +
                    "       b.MARKTYPE_NEW,\n" +
                    "       a.tm_style,\n" +
                    "       b.JOBID,\n" +
                    "       \"0\" as isadd\n" +
                    "FROM\n" +
                    "  (SELECT *\n" +
                    "   FROM a_dc_ep_ods.zs_t_trademarkinfo\n" +
                    "   WHERE CREDITCODE IS NOT NULL\n" +
                    "     AND CREDITCODE !='') b\n" +
                    "LEFT JOIN\n" +
                    "  a_dc_ep_ods.t_eci_company d on upper(trim(d.credit_code)) = upper(trim(b.CREDITCODE))\n" +
                    "     AND d.credit_code IS NOT NULL and d.credit_code!=''\n" +
                    "INNER JOIN\n" +
                    "  (SELECT m.*,n.credit_code\n" +
                    "   FROM a_dc_ep_dwi.t_tm m\n" +
                    "INNER JOIN\n" +
                    "   a_dc_ep_ods.t_eci_company n\n" +
                    "ON m.company_id = n.company_id\n" +
                    "   WHERE n.credit_code IS NOT NULL\n" +
                    "     AND n.credit_code!='') a ON upper(trim(a.credit_code)) = upper(trim(b.CREDITCODE)) and a.reg_no = b.MARKCODEKEY and a.int_cls = b.UNIONTYPECODE").union(spark.sql("SELECT a.id,\n" +
                    "       a.company_id,\n" +
                    "       d.company_name,\n" +
                    "       a.tm_id,\n" +
                    "       b.MARKNAME,\n" +
                    "       b.MARKCODEKEY,\n" +
                    "       b.APPERSON,\n" +
                    "       a.applicant_en,\n" +
                    "       a.applicant_cn_key_no,\n" +
                    "       case when b.ISINVALID='有效' then '1'" +
                    "            when b.ISINVALID='无效' then '2'" +
                    "            when b.ISINVALID is null then a.status" +
                    "            else '5' end as status,\n" +
                    "       b.UNIONTYPECODE,\n" +
                    "       b.UNIONTYPE,\n" +
                    "       b.APPDATE,\n" +
                    "       a.agent,\n" +
                    "       a.valid_period,\n" +
                    "       a.announcement_issue,\n" +
                    "       a.announcement_date,\n" +
                    "       a.reg_issue,\n" +
                    "       a.reg_date,\n" +
                    "       a.flow_status,\n" +
                    "       a.flow_status_desc,\n" +
                    "       a.flow_items,\n" +
                    "       a.list_group_items,\n" +
                    "       a.address_cn,\n" +
                    "       a.address_en,\n" +
                    "       a.has_image,\n" +
                    "       a.image_url,\n" +
                    "       a.guo_ji_zhu_ce_date,\n" +
                    "       a.hou_qi_zhi_ding_date,\n" +
                    "       a.you_xian_quan_date,\n" +
                    "       a.update_time,\n" +
                    "       b.MARKTYPE_NEW,\n" +
                    "       a.tm_style,\n" +
                    "       b.JOBID,\n" +
                    "       \"0\" AS isadd\n" +
                    "FROM\n" +
                    "  (SELECT *\n" +
                    "   FROM a_dc_ep_ods.zs_t_trademarkinfo\n" +
                    "   WHERE regno IS NOT NULL\n" +
                    "     AND regno !=''\n" +
                    "     AND (CREDITCODE IS NULL\n" +
                    "          OR CREDITCODE = '')) b\n" +
                    "LEFT JOIN\n" +
                    "  a_dc_ep_ods.t_eci_company d on upper(trim(d.no)) = upper(trim(b.regno))\n" +
                    "     AND d.no IS NOT NULL and d.no!=''\n" +
                    "INNER JOIN\n" +
                    "  (SELECT m.*,n.no\n" +
                    "   FROM a_dc_ep_dwi.t_tm m\n" +
                    "INNER JOIN\n" +
                    "   a_dc_ep_ods.t_eci_company n\n" +
                    "ON m.company_id = n.company_id\n" +
                    "   WHERE n.no IS NOT NULL\n" +
                    "     AND n.no!='') a ON upper(trim(a.no)) = upper(trim(b.regno)) and a.reg_no = b.MARKCODEKEY and a.int_cls = b.UNIONTYPECODE"));

            update_data.createOrReplaceTempView("tmp_t_tm");
            spark.sql("insert overwrite table a_dc_ep_incr.t_tm select * from tmp_t_tm");
            //获取中数增量数据表中的新增数据（left join）
            Dataset<Row> insert_data = spark.sql("SELECT b.RECORD_ID,\n" +
                    "       d.company_id,\n" +
                    "       d.company_name,\n" +
                    "       '' as tm_id,\n" +
                    "       b.MARKNAME,\n" +
                    "       b.MARKCODEKEY,\n" +
                    "       b.APPERSON,\n" +
                    "       '' as applicant_en,\n" +
                    "       '' as applicant_cn_key_no,\n" +
                    "       case when b.ISINVALID='有效' then '1'" +
                    "            when b.ISINVALID='无效' then '2'" +
                    "            when b.ISINVALID is null then a.status" +
                    "            else '5' end as status,\n" +
                    "       b.UNIONTYPECODE,\n" +
                    "       b.UNIONTYPE,\n" +
                    "       b.APPDATE,\n" +
                    "       '' as agent,\n" +
                    "       '' as valid_period,\n" +
                    "       '' as announcement_issue,\n" +
                    "       b.CHECKDATE,\n" +
                    "       '' as reg_issue,\n" +
                    "       b.REGDATE,\n" +
                    "       b.STATUS,\n" +
                    "       '' as flow_status_desc,\n" +
                    "       '' as flow_items,\n" +
                    "       '' as list_group_items,\n" +
                    "       '' as address_cn,\n" +
                    "       '' as address_en,\n" +
                    "       '' as has_image,\n" +
                    "       '' as image_url,\n" +
                    "       '' as guo_ji_zhu_ce_date,\n" +
                    "       '' as hou_qi_zhi_ding_date,\n" +
                    "       '' as you_xian_quan_date,\n" +
                    "       '' as update_time,\n" +
                    "       b.MARKTYPE_NEW,\n" +
                    "       '' as tm_style,\n" +
                    "       b.JOBID,\n" +
                    "       \"1\" as isadd\n" +
                    "FROM\n" +
                    "  (SELECT *\n" +
                    "   FROM a_dc_ep_ods.zs_t_trademarkinfo\n" +
                    "   WHERE CREDITCODE IS NOT NULL\n" +
                    "     AND CREDITCODE !='') b\n" +
                    "LEFT JOIN\n" +
                    "  a_dc_ep_ods.t_eci_company d on upper(trim(d.credit_code)) = upper(trim(b.CREDITCODE))\n" +
                    "     AND d.credit_code IS NOT NULL and d.credit_code!=''\n" +
                    "LEFT JOIN\n" +
                    "  (SELECT m.*,n.credit_code\n" +
                    "   FROM a_dc_ep_dwi.t_tm m\n" +
                    "INNER JOIN\n" +
                    "   a_dc_ep_ods.t_eci_company n\n" +
                    "ON m.company_id = n.company_id\n" +
                    "   WHERE n.credit_code IS NOT NULL\n" +
                    "     AND n.credit_code!='') a ON upper(trim(a.credit_code)) = upper(trim(b.creditcode)) and a.reg_no = b.MARKCODEKEY and a.int_cls = b.UNIONTYPECODE\n" +
                    "WHERE a.credit_code IS NULL").union(spark.sql("SELECT b.RECORD_ID,\n" +
                    "       d.company_id,\n" +
                    "       d.company_name,\n" +
                    "       '' as tm_id,\n" +
                    "       b.MARKNAME,\n" +
                    "       b.MARKCODEKEY,\n" +
                    "       b.APPERSON,\n" +
                    "       '' as applicant_en,\n" +
                    "       '' as applicant_cn_key_no,\n" +
                    "       case when b.ISINVALID='有效' then '1'" +
                    "            when b.ISINVALID='无效' then '2'" +
                    "            when b.ISINVALID is null then a.status" +
                    "            else '5' end as status,\n" +
                    "       b.UNIONTYPECODE,\n" +
                    "       b.UNIONTYPE,\n" +
                    "       b.APPDATE,\n" +
                    "       '' as agent,\n" +
                    "       '' as valid_period,\n" +
                    "       '' as announcement_issue,\n" +
                    "       b.CHECKDATE,\n" +
                    "       '' as reg_issue,\n" +
                    "       b.REGDATE,\n" +
                    "       b.STATUS,\n" +
                    "       '' as flow_status_desc,\n" +
                    "       '' as flow_items,\n" +
                    "       '' as list_group_items,\n" +
                    "       '' as address_cn,\n" +
                    "       '' as address_en,\n" +
                    "       '' as has_image,\n" +
                    "       '' as image_url,\n" +
                    "       '' as guo_ji_zhu_ce_date,\n" +
                    "       '' as hou_qi_zhi_ding_date,\n" +
                    "       '' as you_xian_quan_date,\n" +
                    "       '' as update_time,\n" +
                    "       b.MARKTYPE_NEW,\n" +
                    "       '' as tm_style,\n" +
                    "       b.JOBID,\n" +
                    "       \"1\" AS isadd\n" +
                    "FROM\n" +
                    "  (SELECT *\n" +
                    "   FROM a_dc_ep_ods.zs_t_trademarkinfo\n" +
                    "   WHERE regno IS NOT NULL\n" +
                    "     AND regno !=''\n" +
                    "     AND (CREDITCODE IS NULL\n" +
                    "          OR CREDITCODE = '')) b\n" +
                    "LEFT JOIN\n" +
                    "  a_dc_ep_ods.t_eci_company d on upper(trim(d.no)) = upper(trim(b.regno))\n" +
                    "     AND d.no IS NOT NULL and d.no!=''\n" +
                    "LEFT JOIN\n" +
                    "  (SELECT m.*,n.no\n" +
                    "   FROM a_dc_ep_dwi.t_tm m\n" +
                    "INNER JOIN\n" +
                    "   a_dc_ep_ods.t_eci_company n\n" +
                    "ON m.company_id = n.company_id\n" +
                    "   WHERE n.no IS NOT NULL\n" +
                    "     AND n.no!='') a ON upper(trim(a.no)) = upper(trim(b.regno)) and a.reg_no = b.MARKCODEKEY and a.int_cls = b.UNIONTYPECODE\n" +
                    "WHERE a.no IS NULL"));
            insert_data.createOrReplaceTempView("tmp_t_tm1");
            spark.sql("insert into table a_dc_ep_incr.t_tm select * from tmp_t_tm1");
            //获取中数删除数据表中的数据
            Dataset<Row> delete_data = spark.sql("SELECT a.id,\n" +
                    "       a.company_id,\n" +
                    "       a.company_name,\n" +
                    "       a.tm_id,\n" +
                    "       a.name,\n" +
                    "       a.reg_no,\n" +
                    "       a.applicant_cn,\n" +
                    "       a.applicant_en,\n" +
                    "       a.applicant_cn_key_no,\n" +
                    "       a.status,\n" +
                    "       a.int_cls,\n" +
                    "       a.int_cls_desc,\n" +
                    "       a.app_date,\n" +
                    "       a.agent,\n" +
                    "       a.valid_period,\n" +
                    "       a.announcement_issue,\n" +
                    "       a.announcement_date,\n" +
                    "       a.reg_issue,\n" +
                    "       a.reg_date,\n" +
                    "       a.flow_status,\n" +
                    "       a.flow_status_desc,\n" +
                    "       a.flow_items,\n" +
                    "       a.list_group_items,\n" +
                    "       a.address_cn,\n" +
                    "       a.address_en,\n" +
                    "       a.has_image,\n" +
                    "       a.image_url,\n" +
                    "       a.guo_ji_zhu_ce_date,\n" +
                    "       a.hou_qi_zhi_ding_date,\n" +
                    "       a.you_xian_quan_date,\n" +
                    "       a.update_time,\n" +
                    "       a.tm_type,\n" +
                    "       a.tm_style,\n" +
                    "       a.dates,\n" +
                    "       '-1' AS isadd\n" +
                    "FROM\n" +
                    "  (SELECT *\n" +
                    "   FROM a_dc_ep_dwi.t_tm) a\n" +
                    "INNER JOIN\n" +
                    "  (SELECT m.*\n" +
                    "   FROM a_dc_ep_ods.zs_t_trademarkinfo_del m\n" +
                    "   LEFT JOIN a_dc_ep_ods.zs_t_trademarkinfo n ON m.MARKCODEKEY = n.MARKCODEKEY\n" +
                    "   WHERE n.MARKCODEKEY IS NULL) b ON upper(trim(a.reg_no)) = upper(trim(b.MARKCODEKEY))");
            delete_data.createOrReplaceTempView("tmp_t_tm2");
            spark.sql("insert into table a_dc_ep_incr.t_tm select * from tmp_t_tm2");
        }else {
            spark.sql("TRUNCATE TABLE a_dc_ep_incr.t_tm");
            System.out.println("中数本期数据涉及该表的数据为空。。。。。。");
        }

        spark.stop();
    }
}
